package com.pw.study.flink.window

import com.alibaba.fastjson.JSONObject
import com.pw.study.flink.source.StudentSource
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

/**
 * @Desc: 学生的窗口学习
 * @author: future
 * @since: 2022/6/3-11:21
 */
object StudentTask {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val ds: DataStream[JSONObject] = env.addSource(new StudentSource)

    val dsWm: DataStream[JSONObject] = ds.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[JSONObject](Time.seconds(3)) {
      override def extractTimestamp(t: JSONObject): Long = {
        val ts: Long = t.getLongValue("ts")
        println("ts=> " + ts)
        ts
      }
    })
    val dsWindow: DataStream[JSONObject] = dsWm.keyBy(x => {
      val key: String = x.getString("student")
      println(s"key=> $key")
      key
    }).window(TumblingEventTimeWindows.of(Time.seconds(20)))
      .process(new ProcessWindowFunction[JSONObject, JSONObject, String, TimeWindow] {

        var lastTime: ValueState[Long] = _

        override def process(key: String, context: Context, elements: Iterable[JSONObject], out: Collector[JSONObject]): Unit = {
          println("window:: key " + key + s" time: ${context.window.getStart} ==> ${context.window.getEnd} == duration: ${context.window.getEnd - context.window.getStart}" + " size: " + elements.size)
          val arr: Array[JSONObject] = elements.toArray.sortBy(x => x.getLong("ts"))
          for(i<- 0 until(arr.length)){
            val obj: JSONObject = arr(i)
            val stateTs: Long = lastTime.value()
            val cur: Long = obj.getLong("ts")
            if(cur - stateTs>0){
              println(s" hello:>> $cur ===> $stateTs ==$obj")
              lastTime.update(cur)
            }
          }
        }

        override def open(parameters: Configuration): Unit = {
          lastTime = getRuntimeContext.getState(new ValueStateDescriptor[Long]("dateState", classOf[Long]))

        }
      })
    dsWindow.print("window:>")
    env.execute("StudentTask")
  }
}
